package org.example.apitest.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.example.apitest.beans.SensorReading;

public class StateTest3KeyedStateApplicationCase {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(100);

        // 从socket中读取数据
        DataStream<String> inputStream = env.socketTextStream("localhost", 7777);

        // 转换为SensorReading类型
        DataStream<SensorReading> dataStream = inputStream
                .map((MapFunction<String, SensorReading>) SensorReading::new);

        // 定义一个有状态的map操作,统计当前分区数据个数
        SingleOutputStreamOperator<Tuple3<String, Double, Double>> resultStream = dataStream.keyBy("id")
                .flatMap(new RichFlatMapFunction<SensorReading, Tuple3<String, Double, Double>>() {
                    private Double threshold = 10.0;
                    // 定义状态,保存上一次温度值
                    private ValueState<Double> lastTempState;

                    @Override
                    public void open(Configuration param) {
                        lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-temp", Double.class));
                    }

                    @Override
                    public void flatMap(SensorReading value, Collector<Tuple3<String, Double, Double>> out) throws Exception {
                        // 获取状态
                        Double lastTemp = lastTempState.value();
                        // 如果不为null,则判断两次温度差值
                        if (lastTemp != null) {
                            Double diff = Math.abs(value.getTemperature() - lastTemp);
                            if (diff >= threshold) {
                                out.collect(new Tuple3<>(value.getId(), lastTemp, value.getTemperature()));
                            }
                        }
                        // 更新状态
                        lastTempState.update(value.getTemperature());
                    }

                    @Override
                    public void close() {
                        lastTempState.clear();
                    }
                });

        resultStream.print();

        env.execute();
    }
}
